/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* tcp_listen.h
*
* Asynchronous I/O: tpc listener
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#include <ios>
#include <memory>
#include <astreambuf.h>
#include "xxwin32.h"

namespace async
{
    /// <summary>
    /// Signature of handlers for new connections.
    /// </summary>
    /// <param name="stream">A stream buffer that may be used to interact with the other side of the connection.</param>
    /// <param name="state">A pointer to application-specific state that was passed in when the listener was created.</param>
    /// <returns>An error code. 0 means no error.</returns>
    typedef unsigned long (*tcp_handler)(streambuf stream, void *state);

    /// <summary>
    /// Internal base implementation of the TCP listener. Listeners can be established in several ways:
    ///     a) by passing in a function pointer to a handler function.
    ///     b) by overriding the 'new_connection' virtual method in a derived class.
    /// </summary>
    /// <remarks>
    /// If a derived class allows function pointers to be passed in during construction, ***and**
    /// overrides new_connection, the latter will not be called.
    /// </remarks>
    class basic_tcp_listener
    {
    public:
        ~basic_tcp_listener()
        {
            if ( m_state != nullptr ) 
                delete m_state;
        }

        /// <summary>
        /// Create a basic_tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number or service name.</param>
        static pplx::task<basic_tcp_listener *> create(const std::string &host, const std::string &service) 
        {
            auto res = pplx::task_completion_event<basic_tcp_listener *>(); 
            if ( _listen_tcp_str(new _tcpstream_callback_create(res), host.c_str(), service.c_str()) )
                return pplx::task<basic_tcp_listener *>(res);
            else
                return pplx::create_value<basic_tcp_listener *>(nullptr);
        }

        /// <summary>
        /// Open the listener to accept new connections.
        /// </summary>
        /// <returns>True if the opening succeeded, false otherwise.</returns>
        pplx::task<bool> open()
        {
            auto res = pplx::task_completion_event<bool>();
            if ( _open_listener(m_info, new _tcpstream_callback_open(res)) )
                return pplx::task<bool>(res);
            else
                return pplx::create_value(false);
        }

        /// <summary>
        /// Close the listener, i.e. stop accepting new connections. Already open connections are not
        /// affected.
        /// </summary>
        /// <returns>True if it succeeded, false otherwise.</returns>
        pplx::task<bool> close()
        {
            auto res = pplx::task_completion_event<bool>();
			auto result = pplx::task<bool>(res);
            if ( _close_listener(&m_info, new _tcpstream_callback_close(res)) )
                return result;
            else
                return pplx::create_value(false);
        }

    protected:
        /// <summary>
        /// new_connection is called whenever a new TCP connection is established.
        /// </summary>
        /// <param name="stream">A stream buffer that may be used to interact with the other side of the connection.</param>
        /// <returns>An error code. 0 means no error.</returns>
        /// <remarks>The caller will ensure the thread safety, there's no need to take a lock.</remarks>
        virtual unsigned long new_connection(streambuf stream)
        { 
            if ( m_handler != nullptr )
            {
                m_handler(stream, m_state);
            }

            return 0;
        }

    private:

        friend class tcp_listener;
        friend basic_tcp_listener * details::_createtcplst(details::_tcp_info *);

        /// <summary>
        /// Attach a handler function that will be invoked once for each new connection that is established
        /// with the listener.
        /// </summary>
        /// <param name="handler">A pointer to a handler function.</param>
        /// <param name="state">A pointer to application-specific state, which will be passed through to the handler function.</param>
        void _new_connection(tcp_handler handler, void *state)
        {
            if ( m_info == nullptr ) return;

            actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
            m_handler = handler;
            if ( m_state != nullptr ) 
                delete m_state;
            m_state = state;
        }

        basic_tcp_listener(details::_tcp_info *info) : m_info(info), m_handler(nullptr), m_state(nullptr) { }

        class _tcpstream_callback_create : public details::_listener_callback
        {
        public:
            _tcpstream_callback_create(pplx::task_completion_event<basic_tcp_listener *> op) : m_op(op) { }

            virtual void on_listening(basic_tcp_listener *listener)
            {
                m_op.set(listener);
                delete this;
            }

        private:
            pplx::task_completion_event<basic_tcp_listener *> m_op;
        };

        class _tcpstream_callback_open : public details::_listener_callback
        {
        public:
            _tcpstream_callback_open(pplx::task_completion_event<bool> op) : m_op(op) { }

            virtual void on_opened(bool result) 
            {
                m_op.set(result);
            }

            virtual void on_accept(basic_tcp_listener *listener, basic_streambuf *buf)
            {
                if ( listener != nullptr) 
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(listener->m_info->m_lock);
                    listener->new_connection(streambuf(buf));
                }
            }

            virtual void on_closed(bool) { delete this; }

        private:
            pplx::task_completion_event<bool> m_op;
        };

        class _tcpstream_callback_close : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_close(pplx::task_completion_event<bool> op) : m_op(op) { }

            virtual void on_closed(bool result)
            {
                m_op.set(result);
                delete this;
            }

        private:
            pplx::task_completion_event<bool> m_op;
        };


        details::_tcp_info *m_info;

        tcp_handler m_handler; 
        void *m_state;
    };

    class tcp_listener
    {
    public:
        /// <summary>
        /// Default constructor
        /// </summary>
        tcp_listener() {}

        /// <summary>
        /// Copy constructor
        /// </summary>
        tcp_listener(const tcp_listener &other) : m_buffer(other.m_buffer) { }

        /// <summary>
        /// Assignment
        /// </summary>
        tcp_listener &operator =(const tcp_listener &other) { m_buffer = other.m_buffer; return *this; }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number or service name.</param>
        static pplx::task<tcp_listener> create(
            const std::string &host, 
            const std::string &service)
        {
            auto bfb = basic_tcp_listener::create(host, service);
            return bfb.then([](pplx::task<basic_tcp_listener *> op) -> tcp_listener
                {
                    return tcp_listener(op.get());
                });
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number or service name.</param>
        static pplx::task<tcp_listener> create(
            const std::wstring &hostname, 
            const std::wstring &service)
        {
            std::string host(utilities::conversions::utf16_to_utf8(std::wstring(hostname)));
            std::string port(utilities::conversions::utf16_to_utf8(std::wstring(service)));
            return create(host, port);
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="port">A port number.</param>
        static pplx::task<tcp_listener> create(
            const std::string &host, 
            int port)
        {
            std::stringstream ss;
            ss << port;
            return create(host, ss.str());
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="port">A port number.</param>
        static pplx::task<tcp_listener> create(
            const std::wstring &host, 
            int port)
        {
            std::wstringstream ss;
            ss << port;
            return create(host, ss.str());
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number or service name.</param>
        /// <param name="handler">A function pointer to call when </param>
        /// <param name="state">A pointer to state that will be passed through to all handlers</param>
        template<typename Functor>
        static pplx::task<tcp_listener> create(
            const std::string &host, 
            const std::string &service, 
            Functor            func)
        {
            auto res = create(host, service);
            return res.then(
                [=](pplx::task<tcp_listener> op) -> tcp_listener
                {
                    auto listener = op.get();
                    listener.on_new_connection(func);
                    listener.open().get();
                    return listener;
                });
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number or service name.</param>
        /// <param name="handler">A function pointer to call when </param>
        /// <param name="state">A pointer to state that will be passed through to all handlers</param>
        template<typename Functor>
        static pplx::task<tcp_listener> create(
            const std::wstring &hostname, 
            const std::wstring &service, 
            Functor            func)
        {
            std::string host(utilities::conversions::utf16_to_utf8(std::wstring(hostname)));
            std::string port(utilities::conversions::utf16_to_utf8(std::wstring(service)));
            return create(host, port, func);
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number.</param>
        /// <param name="handler">A function pointer to call when </param>
        /// <param name="state">A pointer to state that will be passed through to all handlers</param>
        template<typename Functor>
        static pplx::task<tcp_listener> create(
            const std::string &host, 
            int port, 
            Functor            func)
        {
            std::stringstream ss;
            ss << port;
            return create(host, ss.str(), func);
        }

        /// <summary>
        /// Create a tcp_listener asynchronously
        /// </summary>
        /// <param name="host">The name/address of the host. An empty string implies 'localhost'.</param>
        /// <param name="service">A port number.</param>
        /// <param name="handler">A function pointer to call when </param>
        /// <param name="state">A pointer to state that will be passed through to all handlers</param>
        template<typename Functor>
        static pplx::task<tcp_listener> create(
            const std::wstring &host, 
            int port, 
            Functor            func)
        {
            std::wstringstream ss;
            ss << port;
            return create(host, ss.str(), func);
        }

        /// <summary>
        /// Open the listener to accept new connections.
        /// </summary>
        /// <returns>True if the opening succeeded, false otherwise.</returns>
        pplx::task<bool> open()
        {
            if ( m_buffer )
                return m_buffer->open();
            else
                return pplx::create_value(false);
        }

        /// <summary>
        /// Close the listener, i.e. stop accepting new connections. Already open connections are not
        /// affected.
        /// </summary>
        /// <returns>True if it succeeded, false otherwise.</returns>
        pplx::task<bool> close()
        {
            if ( m_buffer )
                return m_buffer->close();
            else
                return pplx::create_value(false);
        }

        /// <summary>
        /// Attach a handler function that will be invoked once for each new connection that is established
        /// with the listener.
        /// </summary>
        /// <param name="handler">A callable object that will be called for each new connection</param>
        /// <remarks>The functor is given a tcpstreambuf as its only argument.</remarks>
        template<typename Functor>
        void on_new_connection(Functor handler)
        {
            if ( m_buffer ) 
            {
                HandlerHelper<Functor> *helper = CreateHelper(handler);
                m_buffer->_new_connection(HandlerHelper<Functor>::Handler, helper);
            }
        }

    private:
        
        template<typename Functor>
        class HandlerHelper
        {
        public:
            HandlerHelper(Functor func) : m_func(func) {}

        private:
            friend class tcp_listener;

            static unsigned long Handler(streambuf stream, void *state)
            {
                auto _this = (HandlerHelper<Functor> *)state;
                _this->m_func(stream);
                return 0;
            }

            Functor m_func;
        };

        template<typename Functor>
        HandlerHelper<Functor> *CreateHelper(Functor func)
        {
            return new HandlerHelper<Functor>(func);
        }
        
        tcp_listener(basic_tcp_listener *buffer) : m_buffer(buffer) {}

        std::shared_ptr<basic_tcp_listener> m_buffer;
    };
}